package com.tdy.cdc.app.dwd;

import com.tdy.cdc.app.BaseSqlApp;
import com.tdy.cdc.common.Constant;
import com.tdy.cdc.util.FlinkSinkUtil;
import com.tdy.cdc.util.FlinkSourceUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

/**
 * @author NanHuang
 * @Date 2023/3/10
 */
public class App01_DwdMhsEoMr extends BaseSqlApp {
    public static void main(String[] args) {
        new App01_DwdMhsEoMr().init(
                5001,
                "App1_DwdMhsEoMr",
                2
        );
    }

    @Override
    protected void invoke(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv) {
        // 1 创建源表
        createSourceTables(tableEnv);
        // 2 join
        joinTables(tableEnv);
        // 3 写入Kafka
        writeToKafka(tableEnv);
    }

    private void writeToKafka(StreamTableEnvironment tableEnv) {
        // 创建Kafka映射表
        tableEnv.executeSql("create table target (" +
                "RECORD_ID                      STRING," +
                "HOSPITAL_NAME                  STRING," +
                "HOSPITAL_CODE                  STRING," +
                "AGE_UNIT                       STRING," +
                "AGE                            STRING," +
                "SEX_NAME                       STRING," +
                "SEX_CODE                       STRING," +
                "ID_CARD_NUMBER                 STRING," +
                "PATIENT_VISIT_CATEGORY_CODE    STRING," +
                "PATIENT_VISIT_CATEGORY         STRING," +
                "OUTPATIENT_VISIT_FLOW_ID       STRING," +
                "OUTPATIENT_NO                  STRING," +
                "PATIENT_ID                     STRING," +
                "PATIENT_NAME                   STRING," +
                "DOMAIN_ID                      STRING," +
                "TELECOM                        STRING," +
                "VISIT_DATE                     STRING," +
                "VISIT_TIMES                    STRING," +
                "APPLY_SHEET_NO                 STRING," +
                "BIRTH_DATE                     STRING," +
                "DEPT_ID                        STRING," +
                "DEPT_NAME                      STRING," +
                "AUTHOR_CODE                    STRING," +
                "AUTHOR_NAME                    STRING," +
                "CUSTODIAN_CODE                 STRING," +
                "CUSTODIAN_NAME                 STRING," +
                "RESPONSE_TIME                  STRING," +
                "RESPONSE_CODE                  STRING," +
                "RESPONSE_NAME                  STRING," +
                "ALLERGIC_HISTORY_FLAG          STRING," +
                "ALLERGIC_HISTORY               STRING," +
                "CHIEF_COMPLAINT                STRING," +
                "PRESENT_ILLNESS_HISTORY        STRING," +
                "PAST_ILLNESS_HISTORY           STRING," +
                "PHYSICAL_EXAM                  STRING," +
                "FIRST_VISIT_FLAG               STRING," +
                "PD_NAME                        STRING," +
                "CHINA_MED_OBSERVE_RESULT       STRING," +
                "DISCRIMINATE_BASIS             STRING," +
                "RULES_CODE                     STRING," +
                "RESCUE_START_DATE              STRING," +
                "RESCUE_END_DATE                STRING," +
                "RESCUERS_LIST                  STRING," +
                "PAT_CATEGORY_CODE              STRING," +
                "PAT_CATEGORY                   STRING," +
                "INCOME_OBSERVATORY_DATE        STRING," +
                "EMERGENCY_COURSE               STRING," +
                "NEEDING_ATTENTION              STRING," +
                "DESTINATION_CODE               STRING," +
                "EMERGENCY_RECORD               STRING," +
                "DIAG_CODE                      STRING," +
                "DIAG_NAME                      STRING," +
                "CHINA_DISEASE_CODE             STRING," +
                "CHINA_DISEASE_NAME             STRING," +
                "CHINA_SYNDROME_CODE            STRING," +
                "CHINA_SYNDROME_NAME            STRING," +
                "ORDERS_AUDIT_STAFF_ID          STRING," +
                "ORDERS_AUDIT_TIME              STRING," +
                "ORDERS_AUDIT_STAFF_NAME        STRING," +
                "ORDERS_CANCEL_DOCTOR_ID        STRING," +
                "ORDERS_CANCEL_TIME             STRING," +
                "ORDERS_CANCEL_DOCTOR_NAME      STRING," +
                "ORDERS_EXEC_STAFF_ID           STRING," +
                "ORDERS_EXEC_STAFF_NAME         STRING," +
                "ORDERS_EXEC_TIME               STRING," +
                "ORDER_EXEC_DEPT                STRING," +
                "EXECUTOR_STATUS                STRING," +
                "ORDERS_CREATE_DOCTOR_ID        STRING," +
                "ORDERS_CREATE_TIME             STRING," +
                "ORDERS_CREATE_DEPT_NAME        STRING," +
                "ORDERS_CREATE_DOCTOR_NAME      STRING," +
                "ORDER_CODE                     STRING," +
                "ORDER_CONTENT                  STRING," +
                "ORDERS_COMMENTS                STRING," +
                "ORDER_TYPE                     STRING," +
                "ORDERS_START_TIME              STRING," +
                "ORDERS_END_TIME                STRING," +
                "ASSIST_EXAM_ITEM               STRING," +
                "ASSIST_EXAM_RESULT             STRING," +
                "VISIT_ID                       STRING," +
                "primary key (RECORD_ID) not enforced" +
                ")" +
                FlinkSinkUtil.getUpsertKafkaWith(Constant.TOPIC_DWD_MHS_EO_MR));
        // 导入数据
        tableEnv.executeSql("insert into target select * from join_result");
    }

    private void joinTables(StreamTableEnvironment tableEnv) {
        // 1 设置数据的TTL（暂时30min）
        tableEnv.getConfig().setIdleStateRetention(Duration.ofSeconds(30 * 60));
        // 2 join操作
        Table joinResult = tableEnv.sqlQuery("select " +
                "s.unique_id as RECORD_ID , " +
                "hospital_name as HOSPITAL_NAME , " +
                "hospital_code as HOSPITAL_CODE , " +
                "age_unit as AGE_UNIT , " +
                "age as AGE , " +
                "sex as SEX_NAME , " +
                "sex_code as SEX_CODE , " +
                "identify_no as ID_CARD_NUMBER , " +
                "patient_typecode as PATIENT_VISIT_CATEGORY_CODE , " +
                "patient_type as PATIENT_VISIT_CATEGORY , " +
                "visit_id as OUTPATIENT_VISIT_FLOW_ID , " +
                "out_patient_id as OUTPATIENT_NO , " +
                "patient_id as PATIENT_ID , " +
                "patient_name as PATIENT_NAME , " +
                "s.domain_id as DOMAIN_ID , " +
                "telecom as TELECOM , " +
                "visit_date as VISIT_DATE , " +
                "visit_times as VISIT_TIMES , " +
                "s.appy_id as APPLY_SHEET_NO , " +
                "birth as BIRTH_DATE , " +
                "dept_code as DEPT_ID , " +
                "dept_name as DEPT_NAME , " +
                "author_code as AUTHOR_CODE , " +
                "author_name as AUTHOR_NAME , " +
                "custodian_code as CUSTODIAN_CODE , " +
                "custodian_name as CUSTODIAN_NAME , " +
                "response_time as RESPONSE_TIME , " +
                "response_code as RESPONSE_CODE , " +
                "response_name as RESPONSE_NAME , " +
                "cast(case when allergy_history_mark='1' then allergy_history_mark='true' else allergy_history_mark='false' end as string) as ALLERGIC_HISTORY_FLAG , " +
                "allergy_history as ALLERGIC_HISTORY , " +
                "chief_complaint as CHIEF_COMPLAINT , " +
                "current_history as PRESENT_ILLNESS_HISTORY , " +
                "past_history as PAST_ILLNESS_HISTORY , " +
                "physical_examination as PHYSICAL_EXAM , " +
                "pd_code as FIRST_VISIT_FLAG , " +
                "pd_name as PD_NAME , " +
                "tcm_four_diagnosis as CHINA_MED_OBSERVE_RESULT , " +
                "discriminate_basis as DISCRIMINATE_BASIS , " +
                "rules_code as RULES_CODE , " +
                "rescue_start_date as RESCUE_START_DATE , " +
                "rescue_end_date as RESCUE_END_DATE , " +
                "rescuers_list as RESCUERS_LIST , " +
                "pat_category_code as PAT_CATEGORY_CODE , " +
                "pat_category as PAT_CATEGORY , " +
                "income_observatory_date as INCOME_OBSERVATORY_DATE , " +
                "emergency_course as EMERGENCY_COURSE , " +
                "needing_attention as NEEDING_ATTENTION , " +
                "destination_code as DESTINATION_CODE , " +
                "emergency_record as EMERGENCY_RECORD , " +
                "case when diag_type='西医诊断' then diag_code else '' end as DIAG_CODE , " +
                "case when diag_type='西医诊断' then diag_name else '' end as DIAG_NAME , " +
                "case when diag_type='中医病名' then diag_code else '' end as CHINA_DISEASE_CODE , " +
                "case when diag_type='中医病名' then diag_name else '' end as CHINA_DISEASE_NAME , " +
                "case when diag_type='中医证候' then diag_code else '' end as CHINA_SYNDROME_CODE , " +
                "case when diag_type='中医证候' then diag_name else '' end as CHINA_SYNDROME_NAME , " +
                "audit_code as ORDERS_AUDIT_STAFF_ID , " +
                "audit_date as ORDERS_AUDIT_TIME , " +
                "audit_order as ORDERS_AUDIT_STAFF_NAME , " +
                "cancel_code as ORDERS_CANCEL_DOCTOR_ID , " +
                "cancel_date as ORDERS_CANCEL_TIME , " +
                "cancel_name as ORDERS_CANCEL_DOCTOR_NAME , " +
                "executor_code as ORDERS_EXEC_STAFF_ID , " +
                "executor_name as ORDERS_EXEC_STAFF_NAME , " +
                "executor_date as ORDERS_EXEC_TIME , " +
                "executor_dept as ORDER_EXEC_DEPT , " +
                "executor_status as EXECUTOR_STATUS , " +
                "open_code as ORDERS_CREATE_DOCTOR_ID , " +
                "open_date as ORDERS_CREATE_TIME , " +
                "open_dept as ORDERS_CREATE_DEPT_NAME , " +
                "open_name as ORDERS_CREATE_DOCTOR_NAME , " +
                "order_code as ORDER_CODE , " +
                "order_content as ORDER_CONTENT , " +
                "order_remark as ORDERS_COMMENTS , " +
                "order_type as ORDER_TYPE , " +
                "start_date as ORDERS_START_TIME , " +
                "end_date as ORDERS_END_TIME , " +
                "item_name as ASSIST_EXAM_ITEM , " +
                "result_name as ASSIST_EXAM_RESULT," +
                "visit_id  as VISIT_ID  " +
                "from hdsd00_03_01 s    " +
                "left join tdy_list_diag d on s.unique_id = d.unique_id and s.xds_type = d.xds_type    " +
                "left join tdy_list_order o on s.unique_id = o.unique_id and s.xds_type = o.xds_type    " +
                "left join tdy_list_lexam l on s.unique_id = l.unique_id and s.xds_type = l.xds_type    ");
        tableEnv.createTemporaryView("join_result",joinResult);
    }

    private void createSourceTables(StreamTableEnvironment tableEnv) {
        // hdsd00_03_01
        tableEnv.executeSql("create table hdsd00_03_01 (" +
                "pk                      string," +
                "upload_time             string," +
                "status                  string," +
                "empi                    string," +
                "encounter_id            string," +
                "telecom                 string," +
                "patient_id              string," +
                "patient_domain          string," +
                "visit_date              string," +
                "visit_domain            string," +
                "visit_id                string," +
                "visit_times             string," +
                "unique_id               string," +
                "xds_type                string," +
                "xds_name                string," +
                "effective_time          string," +
                "xds_version             string," +
                "out_patient_id          string," +
                "domain_id               string," +
                "appy_id                 string," +
                "patient_type            string," +
                "identify_no             string," +
                "patient_typecode        string," +
                "patient_name            string," +
                "sex_code                string," +
                "sex                     string," +
                "birth                   string," +
                "age                     string," +
                "age_unit                string," +
                "dept_code               string," +
                "dept_name               string," +
                "hospital_code           string," +
                "hospital_name           string," +
                "write_time              string," +
                "author_code             string," +
                "author_name             string," +
                "custodian_code          string," +
                "custodian_name          string," +
                "response_time           string," +
                "response_code           string," +
                "response_name           string," +
                "allergy_history_mark    string," +
                "allergy_history         string," +
                "chief_complaint         string," +
                "current_history         string," +
                "past_history            string," +
                "physical_examination    string," +
                "pd_code                 string," +
                "pd_name                 string," +
                "tcm_four_diagnosis      string," +
                "discriminate_basis      string," +
                "rules_code              string," +
                "rescue_start_date       string," +
                "rescue_end_date         string," +
                "rescuers_list           string," +
                "pat_category_code       string," +
                "pat_category            string," +
                "income_observatory_date string," +
                "emergency_course        string," +
                "needing_attention       string," +
                "destination_code        string," +
                "emergency_record        string," +
                "check_item              string," +
                "item_result             string" +
                ")" +
                FlinkSourceUtil.getKafkaWith("hdsd00_03_01","App1_DwdMhsEoMr"));
        // tdy_list_diag
        super.readTdyListDiag(tableEnv,"App1_DwdMhsEoMr");
        // tdy_list_order
        super.readTdyListOrder(tableEnv,"App1_DwdMhsEoMr");
        // tdy_list_lexam
        super.readTdyListLexam(tableEnv,"App1_DwdMhsEoMr");
    }
}
